package com.springboot.lock.lock_zk.lock;

import com.springboot.lock.lock_redis.lock.DistributeLock;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

/**
 * @Author: TongRui乀
 * @Date: 2020-08-06 19:12
 * @description：
 */
@Slf4j
@Service
public class ZkDistributeLock implements ZkLock {

    private final String ROOT = "/locks";
    /**
     *  这里可重入锁的设计应该存放 多个节点 表示可重入
     */
    private ThreadLocal<ArrayList<String>> threadLocal = new ThreadLocal<>();

    /**
     * zookeeper 客户端
     */
    private ZooKeeper zk;

    private CountDownLatch zkServerEvent = new CountDownLatch(1);

    {
        try {
            zk = new ZooKeeper("127.0.0.1:2181", 200000, watchedEvent -> {
                if (watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected) {
                    zkServerEvent.countDown();
                }
            });

            zkServerEvent.await();

            log.info("zookeeper client start success...");

            // 判断lock 根节点是否存在
            Stat exists = zk.exists(ROOT, false);

            if (exists == null) {
                zk.create(ROOT, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (IOException | InterruptedException | KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String tryLock(String key, String value, Integer expire) {
        return null;
    }

    @Override
    public String tryLock(String key, String value, Integer expire, long retryMills) {

        try {
            // 是否是重入
            if (threadLocal.get() != null) {
                for (String node : threadLocal.get()) {
                    if(zk.exists(node, false) != null){
                        return node;
                    }
                }
            }
            // 在根节点下创建临时有序的子节点
            String node = zk.create(ROOT + key, value.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            // 获取根节点下所有的子节点
            List<String> children = zk.getChildren(ROOT, false);

            Collections.sort(children);

            children = children.stream().map(child -> ROOT + "/" + child).collect(Collectors.toList());

            if (node.equals(children.get(0))) {
                // 获取到锁
                ArrayList<String> nodes;
                if(threadLocal.get() == null){
                    nodes = new ArrayList<>();
                }else {
                    nodes = threadLocal.get();
                }
                nodes.add(node);
                threadLocal.set(nodes);
                return node;
            }

            /*if (retryMills > 0) {
                // 未获取到锁 获取前一个节点 这里可以确保所有的线程都能根据自己创建子节点的顺序获取到锁
                String preNode = children.get(children.indexOf(node) - 1);

                zk.exists(preNode, watchedEvent -> {
                    if (watchedEvent.getType() == Watcher.Event.EventType.NodeDeleted) {
                        // 监听上一个节点的删除事件

                    }
                });
            }*/

            // 没有获取到锁 删掉自己创建的节点
            zk.delete(node, -1);


        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void unLock(String key) {

        try {
            ArrayList<String> nodes = threadLocal.get();
            if(nodes != null){
                zk.delete(key, -1);
                nodes.remove(key);
                threadLocal.set(nodes);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }

        log.info("{} 释放锁 :{}", Thread.currentThread().getName(), key);

        threadLocal.remove();

    }

}
